#include "epoll_impl.h"
#include "srv_asynch_man.h"

#include<pthread.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <errno.h>
#include <iostream>
using namespace std;
#include <string.h>
#define MAX_EVENTS 10000//10000 最大个客户段链接数
#define MAXLINE 2048
Epoll_Impl::Epoll_Impl(Srv_Asynch_Man *pAsynch_man):m_pAsynch_Man(pAsynch_man)
{
    h_threadId = 0;

    g_epollFd  = 0;

    m_listenFd = 0;

    memset(fd_pipe,0,sizeof(fd_pipe));
}
Epoll_Impl::~Epoll_Impl()
{
    close();
}

int Epoll_Impl::setnonblocking(int sockfd)
{
    if (fcntl(sockfd, F_SETFL, fcntl(sockfd, F_GETFD, 0)|O_NONBLOCK) == -1) {
        return -1;
    }
    return 0;
}
void Epoll_Impl::mod_fd(int fd, uint32_t fig )//将fd添加到事件表epollfd
{
    struct epoll_event event;
    event.data.fd = fd;
    event.events = fig;//采用EPOLLONESHOT事件避免了同一事件被多次触发，因为一个事件只被触发一次且需要重置事件才能侦听下次是否发生
    if (epoll_ctl(g_epollFd, EPOLL_CTL_MOD, fd, &event) < 0)
    {
       fprintf(stderr, "epoll mod error: fd=%d\n", m_listenFd);
    }
}


void Epoll_Impl::add_fd(int fd, uint32_t fig )//将fd添加到事件表epollfd
{
    struct epoll_event event;
    event.data.fd = fd;
    event.events = fig;//采用EPOLLONESHOT事件避免了同一事件被多次触发，因为一个事件只被触发一次且需要重置事件才能侦听下次是否发生
    if (epoll_ctl(g_epollFd, EPOLL_CTL_ADD, fd, &event) < 0)
    {
       fprintf(stderr, "epoll add error: fd=%d\n", m_listenFd);
    }
}

void Epoll_Impl::del_fd(int fd)
{
    struct epoll_event event;
    event.data.fd = fd;
     if (epoll_ctl(g_epollFd, EPOLL_CTL_DEL, fd, &event) < 0)
    {
       fprintf(stderr, "epoll del error: fd=%d\n", m_listenFd);
    }
}

void Epoll_Impl::epoll_Init(const char *pAddr,int nPort)
{
    //创建同步通知管道
    if (pipe(fd_pipe) < 0)
     {
         printf("Unable to create pipe!\n");
         return ;
     }

    //创建socket
    m_listenFd = socket(AF_INET, SOCK_STREAM, 0);

    if (setnonblocking(m_listenFd) < 0)
    {
        perror("setnonblock error");
    }

    sockaddr_in sin;
    bzero(&sin, sizeof(sin));
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = inet_addr(pAddr);//INADDR_ANY;//
    sin.sin_port = htons(nPort);
    bind(m_listenFd, (const sockaddr*)&sin, sizeof(sin));
    listen(m_listenFd, 5);

    g_epollFd = epoll_create(MAX_EVENTS);

    //add_fd(m_listenFd,EPOLLIN | EPOLLET|EPOLLONESHOT);
    //add_fd(fd_pipe[0],EPOLLIN | EPOLLET|EPOLLONESHOT);

    add_fd(m_listenFd,EPOLLIN | EPOLLET);
    add_fd(fd_pipe[0],EPOLLIN | EPOLLET);
    //创建线程
    pthread_t threadID;

    int  err = pthread_create(&threadID, NULL, svc_run, (void*)this);

    h_threadId = (int)threadID;//记录线程ID

}
int Epoll_Impl::svc()
{
    cout << "svc ...bigen" << endl;
    struct epoll_event *events = new epoll_event[MAX_EVENTS];

    while(true)
    {
         /* 等待有事件发生 */
        int nfds = epoll_wait(g_epollFd, events, MAX_EVENTS, -1);   //得到活动的套接字
        //cout << ".............." << endl;
        if (nfds == -1)
        {
            cout << "if (nfds == -1)" << endl;
            if(!g_epollFd)
            {
                 cout << "tiao" << endl;
                break;
            }


             perror("epoll_wait");
             continue;
        }
         //cout << "处理所有事件" << endl;
        /* 处理所有事件 */
        for (int n = 0; n < nfds; ++n)
        {
            if(events[n].data.fd == fd_pipe[0])
            {
                ::close(g_epollFd);
                g_epollFd  =  0;

                ::close(m_listenFd);
                 m_listenFd  =  0;

                ::close(fd_pipe[1]);
                ::close(fd_pipe[0]);
                 cout << "tiao1" << endl;
                break;

            }else if (events[n].data.fd == m_listenFd)
            {
                connect_prc();  //

                //mod_fd(events[n].data.fd,EPOLLIN | EPOLLET|EPOLLONESHOT);

            }else if(events[n].events & EPOLLIN)//如果是已经连接的用户，并且收到数据，那么进行读入。
            {
                //read_prc(events[n]); //
                if(m_pAsynch_Man)
                    m_pAsynch_Man->post_socket_read(events[n].data.fd);
                //mod_fd(events[n].data.fd,EPOLLIN | EPOLLET|EPOLLONESHOT);

            }else if(events[n].events & EPOLLOUT) // 如果有数据发送
            {
               // cout << "EPOLLOUT" << endl;
                if(m_pAsynch_Man)
                    m_pAsynch_Man->post_socket_send(events[n].data.fd); //  post 通知 write_queue 可以发送数据了
                //send_prc(events[n]);
                //mod_fd(events[n].data.fd,EPOLLIN | EPOLLET|EPOLLONESHOT);
            }
        }
    }
    delete []events;//回收内存

    cout << "svc ...exit" << endl;
    return 0;
}

void Epoll_Impl::read_passive_prc(int sockfd, string & strbuf)
{
    char buf[MAXLINE];
    while(true)
    {
        // 确保sockfd是nonblocking的
        int  recvNum = recv(sockfd, buf, MAXLINE, 0);
        if(recvNum < 0)
        {
            if(errno == EAGAIN)//读完了
            {
               // 由于是非阻塞的模式,所以当errno为EAGAIN时,表示当前缓冲区已无数据可读
               // 在这里就当作是该次事件已处理处.
               break;
            }
            else if (errno == EINTR)// 被信号中断
            {
               continue;
            }
            else//其他不可弥补的错误
            {
                if(m_pAsynch_Man)
                    m_pAsynch_Man->delete_service_handler(sockfd);

               destroy_connection_prc(sockfd);

               cout << "unrecovable error1\n";
               break;
            }
        }else if(recvNum == 0)   //对方正常关闭套接字
        {
            cout << "unrecovable error2 \n";

            if(m_pAsynch_Man)
                m_pAsynch_Man->delete_service_handler(sockfd);

            destroy_connection_prc(sockfd);

            break;
        }else
        {
            strbuf.append(buf,recvNum);

            if(recvNum == MAXLINE)//读完了
            {
                continue;
            }
            break;
        }
    }
}

void Epoll_Impl::destroy_connection_prc(int sockfd)
{
    del_fd(sockfd);  //删除epoll管理的套接字

   ::close(sockfd);
}

bool Epoll_Impl::send_passive_prc(int sockfd, string & strbuf)
{
    //cout << "send_passive_prc" << endl;
    while(true)
    {
        int sendLen =  send(sockfd, strbuf.data(), strbuf.size(), 0);
        if (sendLen < 0)
        {
           if (errno == EAGAIN)     //发送缓冲区满了
           {
               break;  //发送缓冲区满了 跳出循环

           }else if (errno == EINTR)// 被信号中断
           {
              continue;
           }
           else//其他不可弥补的错误
           {

               if(m_pAsynch_Man)
                   m_pAsynch_Man->delete_service_handler(sockfd);

              destroy_connection_prc(sockfd);

              cout << "unrecovable error2 " << endl;
              break;   //错误 关闭套接字 跳出循环
           }
        }else if (sendLen == 0)// 这里表示对端的socket已正常关闭.
        {
            cout << "unrecovable error4" << endl;

            if(m_pAsynch_Man)
                m_pAsynch_Man->delete_service_handler(sockfd);

            destroy_connection_prc(sockfd);

            break;      //关闭套接字 跳出循环
        }else
        {
             if(sendLen < (int)strbuf.size())
             {
                 strbuf = string(strbuf.begin()+ sendLen,strbuf.end());//保存没有发送的部分

             }else
             {
                 strbuf = "" ;  //清空 //刚好发完

                 mod_fd(sockfd,EPOLLIN | EPOLLET);//发送成功 暂时没有需要发送的数据 就得去掉out事件

                 return true;
             }
             break;     // 跳出循环
        }
    }
    return false;
}

bool Epoll_Impl::send_active_prc(int sockfd, string & strbuf)
{
    //cout << "send_active_prc" << endl;
    while(true)
    {
        int sendLen =  send(sockfd, strbuf.data(), strbuf.size(), 0);
        if (sendLen < 0)
        {
           if (errno == EAGAIN)     //发送缓冲区满了
           {
               mod_fd(sockfd,EPOLLOUT |EPOLLET);//out事件 EPOLLIN |

               return true;

              // break;  //发送缓冲区满了 跳出循环

           }else if (errno == EINTR)// 被信号中断
           {
              continue;
           }
           else//其他不可弥补的错误
           {
               if(m_pAsynch_Man)
                   m_pAsynch_Man->delete_service_handler(sockfd);

               destroy_connection_prc(sockfd);

              cout << "unrecovable error5" << endl;

              break;   //错误 关闭套接字 跳出循环
           }
        }else if (sendLen == 0)// 这里表示对端的socket已正常关闭.
        {
            cout << "unrecovable error6" << endl;

            if(m_pAsynch_Man)
                m_pAsynch_Man->delete_service_handler(sockfd);

            destroy_connection_prc(sockfd);

            break;      //关闭套接字 跳出循环
        }else
        {
             if(sendLen < (int)strbuf.size())
             {
                 strbuf = string(strbuf.begin()+ sendLen,strbuf.end());//保存没有发送的部分

                 mod_fd(sockfd,EPOLLOUT |EPOLLET);//out事件 EPOLLIN |

                 return true;
             }else
             {
                 strbuf = "" ;  //清空 //刚好发完
             }
             break;
        }
    }
    return false;
}

void Epoll_Impl::connect_prc()
{
    struct sockaddr_in clientaddr;

    socklen_t socklen= sizeof(struct sockaddr);

    int  connfd = 0;
    do
    {
         connfd = accept(m_listenFd, (struct sockaddr *)&clientaddr,&socklen);

         if (connfd < 0)
         {
             // perror("accept error");
              break ;
         }

         if (setnonblocking(connfd) < 0)
         {
            perror("setnonblocking error");
            break ;
         }
         if(m_pAsynch_Man)
             m_pAsynch_Man->add_service_handler(connfd);

         //add_fd(connfd,EPOLLIN |EPOLLET|EPOLLONESHOT);
         add_fd(connfd,EPOLLIN |EPOLLET);

    }while(true);
}

void * Epoll_Impl::svc_run( void *apParam)
{
    ((Epoll_Impl *)apParam)->svc();
}

int Epoll_Impl::close()
{
    write(fd_pipe[1], "123", 3);

    //std::queue<int>  temp = h_threads;

    //cout << h_threads.size() << endl;

    if(h_threadId)
    {
         pthread_join((pthread_t)h_threadId,NULL);//  wait...
         h_threadId = 0;
    }
//    while(!temp.empty())
//    {
//        cout << "xxxxxxxxxxxxxxxxxxxxxx-pthread_join" << endl;
//        pthread_join((pthread_t)temp.front(),NULL);//  wait...
//        temp.pop();
//    }

    cout << "xxxxxxxxxxxxxxxxxxxxxxxxxx" << endl;
//    while(!h_threads.empty())
//          h_threads.pop();

    return 1;
}
